#include <boost/format.hpp>

#include "WebSocket.h"
#include "trace_worker.h"
#include "ClientManager.h"

using websocketpp::connection_hdl;
using websocketpp::lib::placeholders::_1;
using websocketpp::lib::placeholders::_2;
using websocketpp::lib::bind;

CWebSocket::CWebSocket(CRedisSubscriber &subscriber)
:m_subscriber(subscriber)
{   trace_worker();
    // Initialize Asio Transport
    m_server.init_asio();
    // Register handler callbacks
    m_server.set_open_handler(bind(&CWebSocket::on_open,this,::_1));
    m_server.set_close_handler(bind(&CWebSocket::on_close,this,::_1));
    m_server.set_http_handler(bind(&CWebSocket::on_http,this,::_1));
    m_server.set_message_handler(bind(&CWebSocket::on_message,this,::_1,::_2));

    m_messageThread = boost::thread(boost::bind(&CWebSocket::process_messages, this));
}

CWebSocket::~CWebSocket()
{

}

void CWebSocket::run(boost::uint16_t port)
{
    // listen on specified port
    m_server.listen(port);
    // Start the server accept loop
    m_server.start_accept();
    // Start the ASIO io_service run loop
    try
    {
        m_server.run();
    }
    catch (const std::exception & e)
    {
        std::cout << e.what() << std::endl;
    }
}

void CWebSocket::send(const std::string &openId, const std::string &message)
{
    boost::shared_lock<boost::shared_mutex> sharedLock(m_conMapWrMutex);
    ConMap::iterator conMapIter = m_conMap.find(openId);
    if (conMapIter != m_conMap.end())
    {
        m_server.send(conMapIter->second, message, websocketpp::frame::opcode::text);
    }
}

void CWebSocket::on_open(connection_hdl hdl)
{   trace_worker();
    server::connection_ptr con = m_server.get_con_from_hdl(hdl);    

    std::size_t pos = con->get_resource().find_last_of('/');
    if (pos == std::string::npos)
    {   trace_printf("con->get_resource().find_last_of('/')  %s", con->get_resource().c_str());
        return;
    } 
    std::string openId(con->get_resource().substr(pos+1));
    boost::unique_lock<boost::shared_mutex> uniqueLock(m_conMapWrMutex);  
    ConMap::iterator conMapIter = m_conMap.find(openId);
    if (conMapIter == m_conMap.end())
    {
        m_subscriber.subscribe(openId);
    }
    else
    {
        m_conMap.erase(conMapIter);
    }
    printf("on_openon_openon_openon_open  %s\n", openId.c_str());

    m_conMap.insert(std::make_pair(openId, hdl));
}

void CWebSocket::on_close(connection_hdl hdl)
{   trace_worker();
    server::connection_ptr con = m_server.get_con_from_hdl(hdl);    
    
    std::size_t pos = con->get_resource().find_last_of('/');
    if (pos == std::string::npos)
    {   trace_printf("con->get_resource().find_last_of('/')  %s", con->get_resource().c_str());
        return;
    }
    
    std::string openId(con->get_resource().substr(pos+1));
    boost::unique_lock<boost::shared_mutex> uniqueLock(m_conMapWrMutex); 
    ConMap::iterator conMapIter = m_conMap.find(openId);
    if (conMapIter != m_conMap.end() && m_server.get_con_from_hdl(conMapIter->second) == con)
    {
        printf("on_closeon_closeon_closeon_close  %s\n", openId.c_str());
        m_conMap.erase(conMapIter); 
        m_subscriber.unsubscribe(openId);
    }
}


void CWebSocket::on_message(connection_hdl hdl, server::message_ptr msg)
{
    server::connection_ptr con = m_server.get_con_from_hdl(hdl);    
    std::size_t pos = con->get_resource().find_last_of('/');
    if (pos == std::string::npos)
    {   trace_printf("con->get_resource().find_last_of('/')  %s", con->get_resource().c_str());
        return;
    }
    std::string openId(con->get_resource().substr(pos + 1));
    {
        boost::lock_guard<boost::mutex> guard(m_messageLock);
        m_messageQueue.push(MessageOpenId(msg, openId));
    }
    m_messageCond.notify_one();
}

void CWebSocket::process_messages()
 {
    while(1)
    {
        boost::unique_lock<boost::mutex> lock(m_messageLock);
        while(m_messageQueue.empty()) {
            m_messageCond.wait(lock);
        }
        MessageOpenId messageOpenId = m_messageQueue.front();
        m_messageQueue.pop();
        lock.unlock();


        std::size_t methodPos = messageOpenId.m_message->get_payload().find(':');
        if (methodPos == std::string::npos)
        {
            continue;
        }
        const std::string &method = messageOpenId.m_message->get_payload().substr(0, methodPos);
        const std::string &content = messageOpenId.m_message->get_payload().substr(methodPos + 1);
        printf("process_messages  method.c_str(), content.c_str()  %s  %s\n", method.c_str(), content.c_str());        
        std::string _return;
        CClientManager::instance()->thriftMethod(_return, messageOpenId.m_openId, method, content);
        if (_return == "pong")
        {
            send(messageOpenId.m_openId, _return);    
        } 
    }
}


void CWebSocket::on_http(connection_hdl hdl)
{
    server::connection_ptr con = m_server.get_con_from_hdl(hdl);    
    std::string filename = con->get_resource();
    if (filename.size() == 0 || filename == "/" || filename == "/favicon.ico")
    {
        return;
    }
    std::string queryStr = con->get_uri()->get_query();
    m_httpPara.Parse(queryStr);

    std::string _return;
    CClientManager::instance()->thriftMethodSyn(_return, "", m_httpPara[0], m_httpPara[1]);

    if (m_httpPara[0] == "webview")
    {
        send_html(con, _return);
    }
    else
    {
        con->set_body(_return);
        con->set_status(websocketpp::http::status_code::ok);
    }
}

void CWebSocket::send_html(server::connection_ptr &con, const std::string &content)
{
    con->append_header("Content-Type", "text/html");
    con->append_header("Connection", "close");
    con->append_header("Accept-Ranges", "bytes");
    con->append_header("Content-Length", (boost::format("%d") % content.size()).str());
    con->set_body(content);
    con->set_status(websocketpp::http::status_code::ok);    
}



